package com.rr.learningdemo.flink.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Properties;

/**
 * @author RR
 * @date 2021/7/27 0:02
 */
/*
* 设置时间窗口，每5秒统计一次字符串的出现次数
* */
public class WindowApiTest1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> dataStreamSource = env.socketTextStream("192.168.31.112", 3306);
        dataStreamSource.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                return new Tuple2<>(s, 1);
            }
        }).keyBy(0)
                //滚动窗口，5秒一个，这里两种都是一样的
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5))).sum(1).print();
//                .timeWindow(Time.seconds(5)).sum(1).print();
                //滑动时间窗口，5秒一个窗口，3秒统计一次，即窗口会有交叉的部分
//                .timeWindow(Time.seconds(5), Time.seconds(3)).sum(1).print();
                //滚动计数窗口，当同一个key的来了5次，再做一次统计
//                .countWindow(5).sum(1).print();
                //会话窗口，todo:这个有点搞不懂，以后再来看看，今天好困
                .window(EventTimeSessionWindows.withGap(Time.seconds(5))).sum(1).print();

        env.execute();
    }
}
